/**
 * Copyright (C) 2024-2030 Super BilGen & 超亿时代信息技术股份有限公司.All Rights Reserved.
 */
package cn.com.yang.file;

/**
 * @Model cn.com.yang.thread.IndustrialGradeLineProcessorUTF8
 * @Desc  TODO
 * @author yangyijin
 * @since 2025年5月22日-下午3:23:59
 */
import java.io.*;
import java.nio.*;
import java.nio.channels.FileChannel;
import java.nio.charset.*;
import java.nio.file.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

public class IndustrialGradeLineProcessorUTF8 {

    // 配置参数
    private static final int CPU_CORES = Runtime.getRuntime().availableProcessors();
    private static final int INDEX_BUFFER_SIZE = 256 * 1024;  // 256KB
    private static final int READ_BUFFER_SIZE = 2 * 1024 * 1024; // 2MB
    private static final Charset UTF8 = StandardCharsets.UTF_8;
    private static final AtomicLong globalLineCounter = new AtomicLong(0);

    public static void main(String[] args) throws Exception {
    	 final String inputFile = "F:\\20gb_data.txt";
         final String indexFile = "F:\\20gb_index.idx";

        // 阶段1：构建精准行索引（UTF-8优化）
        if (!Files.exists(Paths.get(indexFile))) {
            buildUTF8Index(inputFile, indexFile);
        }

        // 阶段2：并行处理（UTF-8解码）
        processWithUTF8(inputFile, indexFile);
        System.out.println("总处理行数: " + globalLineCounter.get());
    }

    /** UTF-8 索引构建 */
    private static void buildUTF8Index(String dataFile, String indexFile) throws IOException {
        try (RandomAccessFile raf = new RandomAccessFile(dataFile, "r");
             FileChannel channel = raf.getChannel();
             DataOutputStream indexOut = new DataOutputStream(
                 new BufferedOutputStream(Files.newOutputStream(Paths.get(indexFile))))) {

            ByteBuffer buffer = ByteBuffer.allocate(INDEX_BUFFER_SIZE);
            long filePos = 0;
            boolean newLine = true;
            boolean lastCR = false; // 处理CRLF标记

            while (channel.read(buffer) != -1) {
                buffer.flip();
                while (buffer.hasRemaining()) {
                    byte b = buffer.get();

                    // UTF-8字符边界检查
                    if (newLine && !isUTF8StartByte(b)) {
                        throw new IOException("非UTF-8字符在位置: " + filePos);
                    }

                    // 换行符处理
                    if (b == '\r') {
                        lastCR = true;
                        continue;
                    }

                    if (b == '\n') {
                        if (lastCR) {
                            indexOut.writeLong(filePos - 1); // 记录CR的位置
                        } else {
                            indexOut.writeLong(filePos);
                        }
                        newLine = true;
                    } else {
                        if (newLine) {
                            indexOut.writeLong(lastCR ? filePos - 1 : filePos);
                            newLine = false;
                        }
                    }

                    lastCR = (b == '\r');
                    filePos++;
                }
                buffer.clear();
            }

            // 处理最后一行无换行的情况
            if (!newLine) {
                indexOut.writeLong(filePos);
            }
        }
    }

    /** UTF-8解码处理 */
    private static void processWithUTF8(String dataFile, String indexFile) throws Exception {
        long totalLines = Files.size(Paths.get(indexFile)) / Long.BYTES;
        long linesPerThread = (totalLines + CPU_CORES - 1) / CPU_CORES;

        ExecutorService executor = Executors.newFixedThreadPool(CPU_CORES);
        CompletionService<Long> compService = new ExecutorCompletionService<>(executor);

        try (RandomAccessFile indexRaf = new RandomAccessFile(indexFile, "r")) {
            for (int i = 0; i < CPU_CORES; i++) {
                long startLine = i * linesPerThread;
                long endLine = Math.min(startLine + linesPerThread, totalLines);

                long startByte = getIndexValue(indexRaf, startLine);
                long endByte = (endLine >= totalLines) ? 
                    Files.size(Paths.get(dataFile)) : getIndexValue(indexRaf, endLine);

                compService.submit(() -> 
                    processUTF8Segment(dataFile, startByte, endByte, startLine, endLine),
                    endLine - startLine
                );
            }
        }

        // 验证完整性
        long totalProcessed = 0;
        for (int i = 0; i < CPU_CORES; i++) {
            totalProcessed += compService.take().get();
        }
        executor.shutdown();

        if (totalProcessed != totalLines) {
            throw new IllegalStateException("行数不一致! 预期: " + totalLines + " 实际: " + totalProcessed);
        }
    }

    /** UTF-8分段处理 */
    private static long processUTF8Segment(String filePath, long start, long end, 
                                        long startLine, long endLine) {
        try (RandomAccessFile raf = new RandomAccessFile(filePath, "r");
             FileChannel channel = raf.getChannel()) {

            // 内存映射加速
            MappedByteBuffer buffer = channel.map(
                FileChannel.MapMode.READ_ONLY, start, end - start);
            
            CharsetDecoder decoder = UTF8.newDecoder()
                .onMalformedInput(CodingErrorAction.REPLACE)
                .onUnmappableCharacter(CodingErrorAction.REPLACE);

            CharBuffer charBuffer = decoder.decode(buffer);
            String content = charBuffer.toString();

            // 按行分割
            String[] lines = content.split("\\R"); // 匹配所有换行符
            for (String line : lines) {
            	 System.out.println("line:"+line);
                if (!line.isEmpty()) {
                    globalLineCounter.incrementAndGet();
                    // 示例：处理每行数据
                    // processLine(line);
                }
            }

            return lines.length;
        }catch(IOException ex) {
        	 System.out.println("ex:"+ex);
        }
        return 0;
    }

    /** UTF-8起始字节验证 */
    private static boolean isUTF8StartByte(byte b) {
        int unsignedByte = b & 0xFF;
        return (unsignedByte < 0x80) ||        // 单字节字符
               (unsignedByte >= 0xC2 && unsignedByte < 0xF5 &&  // 多字节字符
                (unsignedByte & 0xC0) == 0xC0); // 检查是否为有效起始字节
    }

    private static long getIndexValue(RandomAccessFile indexRaf, long line) throws IOException {
        indexRaf.seek(line * Long.BYTES);
        return indexRaf.readLong();
    }
}
